package cn.itcast.flink.connector;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lilulu
 */
public class ConnectorKafkaSourceDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        // 2. 数据源-source
//            2.1创建kafkaSource对象，设置属性
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("node1:9092,node2:9092,node3:9092")
                .setTopics("flink-topic")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
//                2.2 添加数据源
        DataStream<String> kafkaStream = env.fromSource(kafkaSource,
                WatermarkStrategy.noWatermarks(), "KafkaSource");
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        kafkaStream.print();
        // 5. 触发执行-execute
        env.execute("ConnectorKafkaSourceDemo");
    }
}